package cn.tedu.batch.transformation

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.scala.ExecutionEnvironment

/**
 * @author Amos
 * @date 2022/5/20
 */

object BatchRebalanceDemo {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.api.scala._

    // 构建数据源
    val source: DataSet[Long] = env.generateSequence(0, 1000000)

    val filterDataSet: DataSet[Long] = source.filter(_ > 0).rebalance()
//    val filterDataSet: DataSet[Long] = source.filter(_ > 0)

    val result = filterDataSet.map(new RichMapFunction[Long,(Int,Long)] {
      override def map(value: Long): (Int, Long) = {
        val taskid = getRuntimeContext.getIndexOfThisSubtask
        (taskid,value)
      }
    })

    result.groupBy(0).reduceGroup(_.size).print()
//    result.print()

  }

}
